package com.qf.cx.simplenms.link.impl;

import java.util.Observable;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.qf.cx.simplenms.link.api.Link;
import com.qf.cx.simplenms.link.api.LinkManager;
import com.qf.cx.simplenms.link.model.OpResultSpec;
import com.qf.cx.simplenms.link.model.OpSpec;
import com.qf.cx.simplenms.link.utils.Utils;

/**
 * Created by linda on 2016/11/12.
 */

public class ReaderLink extends Observable implements Link {
	private static AtomicLong total = new AtomicLong(0);
	private AtomicLong single = new AtomicLong(0);
	private final Logger logger = LoggerFactory.getLogger(this.getClass());
	private final ThreadPoolExecutor es;
	private CanBeStopTask currentTask = NoneTask.INSTANCE;
	private OpSpec savedOpSpec = null;
	private PhysicalLinkMaintainerIF physicalLinkMaintainer;
	private LinkManager.ReaderSource readerSource;
	private int timeSpan = 50;
	private int readTimeout = 5000;
	private int delaytime = 1000;	private int linkErrDelay = 5000;
	private Object lock = new Object();

	ReaderLink(LinkManager.ReaderSource readerSource, PhysicalLinkMaintainerIF physicalLinkMaintainerIF) {
		logger.info("a new ReaderSource:{}", readerSource);
		this.readerSource = readerSource;
		es = new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1), new ThreadFactory() {
			@Override
			public Thread newThread(Runnable r) {
				return new Thread(r, "Reader Link " + readerSource.getReaderIdentifier());
			}
		}, new ThreadPoolExecutor.DiscardOldestPolicy());
		this.physicalLinkMaintainer = physicalLinkMaintainerIF;
	}

	@Override
	public void start(OpSpec opSpec) {
		synchronized (lock) {
			logger.info("reader {} config a new op spec:{}", readerSource, opSpec);
			savedOpSpec = opSpec;
			currentTask.stopTask();
			StartTask newTask = new StartTask();
			es.execute(newTask);
			currentTask = newTask;
		}
	}

	@Override
	public boolean linkOk() {
		return false;
	}

	@Override
	public void stop() {
		logger.info("reader {} stop", readerSource);
		currentTask.stopTask();
	    physicalLinkMaintainer.removeFromAllReaderSources(readerSource);
	}

	class StartTask implements Runnable, CanBeStopTask {
		private final Logger log = LoggerFactory.getLogger(StartTask.class);
		volatile boolean stop = false;

		@Override
		public void stopTask() {
			stop = true;
		}

		@Override
		public void run() {
			while (!stop) {
				final PhysicalLinkIF physicalLinkIF = physicalLinkMaintainer.getLink(readerSource);
				while (!stop) {
					physicalLinkIF.stopOp();
					final SendCommandResultEnum publishOpResult = physicalLinkIF.startOp(savedOpSpec, timeSpan);
					if (publishOpResult == SendCommandResultEnum.FAIL) {
						Utils.sleepms(delaytime);						
						continue;
					} else if (publishOpResult == SendCommandResultEnum.LINK_ERROR) {
						physicalLinkIF.setBreadState();
						Utils.sleepms(linkErrDelay);
						break;
					} else {
						while (!stop) {
							final PeriodOpResult periodOpResult = physicalLinkIF.raiGetPeriodOpResult(readTimeout);

							final SendCommandResultEnum publishGetPeriodOpResultEnum = periodOpResult
									.getSendCommandResultEnum();
							if (publishGetPeriodOpResultEnum == SendCommandResultEnum.FAIL) {
								log.info("StartTask thread get period tags FAILED ");
								continue;
							} else if (publishGetPeriodOpResultEnum == SendCommandResultEnum.LINK_ERROR) {
								log.error("StartTask thread get period tags LINK BREAK ");
								physicalLinkIF.setBreadState();
								break;
							} else {
								final OpResultSpec opResultSpec = periodOpResult.getOpResultSpec();
								if (opResultSpec.getTagResults().isEmpty()) {
									log.debug("StartTask thread get period tags is empty ");
									continue;
								} else {
									single.addAndGet(opResultSpec.getTagResults().size());
									total.addAndGet(opResultSpec.getTagResults().size());
									notifyAllObservers(opResultSpec);
								}
							}
						}
					}
				}
			}
		}
	}

	private void notifyAllObservers(OpResultSpec opResultSpec) {
		setChanged();
		notifyObservers(opResultSpec);
	}

}
